package com.example.micro.canal.client.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;


/**
 * @author Code Farmer
 * @create 2020/10/26 1:37 上午
 */
public class CanalClient {

    private static String SERVER_ADDRESS = "192.168.2.101";

    private static Integer PORT = 11111;

    /**
     * 队列目的地
     * 配置：canal.mq.topic=example
     */
    private static String DESTINATION = "example";

    private static String USERNAME = "";

    private static String PASSWORD = "";

    public static void main(String[] args) {
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(SERVER_ADDRESS, PORT), DESTINATION, USERNAME, PASSWORD
        );
        canalConnector.connect();

        //订阅
        canalConnector.subscribe(".*\\..*");

        //恢复到之前同步的位置
        canalConnector.rollback();

        for (; ; ) {
            //获取100跳数据，但是不做确认，就是这100跳下次还可以获取到
            Message message = canalConnector.getWithoutAck(100);
            long batchId = message.getId();

            if (batchId != -1) {
                System.out.println("batchId-->" + batchId);
                System.out.println(message);
                printEntity(message.getEntries());
            }
        }
    }

    private static void printEntity(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            System.out.println(entry);
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {

            }
        }

    }

}
